package com.boot.rocketmq.producer;


import com.boot.rocketmq.message.DemoTopicMessage;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
public class Demo07Producer {

    private static final String TX_PRODUCER_GROUP="demo07-producer-group";

    private static Map<String, RocketMQLocalTransactionState> STATE_MAP = new HashMap<>();

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public TransactionSendResult sendMessageInTransaction(Integer id) {
        // <1> 创建 Demo07Message 消息
        DemoTopicMessage msg = new DemoTopicMessage();
        msg.setId(id);
        Message<DemoTopicMessage> message = MessageBuilder.withPayload(msg).build();

        // <2> 发送事务消息
        return rocketMQTemplate.sendMessageInTransaction(TX_PRODUCER_GROUP, DemoTopicMessage.TOPIC_07, message,
                id);


    }


    @RocketMQTransactionListener(txProducerGroup = TX_PRODUCER_GROUP)
    public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

        private Logger logger = LoggerFactory.getLogger(getClass());



        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

            String transId =(String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            // ... local transaction process, return rollback, commit or unknown
            logger.info("[executeLocalTransaction][执行本地事务，消息：{} arg：{}]", msg, arg);

            try {
                System.out.println("事务消息成功执行-------");
                STATE_MAP.put(transId,RocketMQLocalTransactionState.COMMIT);
                int a=10/0;
                return RocketMQLocalTransactionState.COMMIT;
            }catch (Exception e){
                e.printStackTrace();
            }
            STATE_MAP.put(transId,RocketMQLocalTransactionState.ROLLBACK);
            return RocketMQLocalTransactionState.UNKNOWN;
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            String transId = (String)msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            // ... check transaction status and return rollback, commit or unknown
            logger.info("[checkLocalTransaction][回查消息：{}],transId:{}", msg,transId);
            //return RocketMQLocalTransactionState.COMMIT;
            return STATE_MAP.get(transId);
        }

    }


}
